Mock SNS publish subscribe fanout in Python

Simulates AWS SNS publish/subscribe with an in-memory topic-to-endpoints dict that fans out messages to all subscribers.

Easy Python 3.9+ Aug 9, 2026 Cloud + Python 13 views 0 copies

Python code

26 lines
Python 3.9+
class SNSMock:
    def __init__(self):
        self.topics = {}

    def create_topic(self, name):
        if name not in self.topics:
            self.topics[name] = []
        return f"arn:aws:sns:us-east-1:123456789012:{name}"

    def subscribe(self, topic_name, endpoint):
        self.topics.setdefault(topic_name, []).append(endpoint)
        return f"subscription:{topic_name}:{endpoint}"

    def publish(self, topic_name, message):
        if topic_name not in self.topics:
            raise ValueError(f"Topic '{topic_name}' does not exist")
        for endpoint in self.topics[topic_name]:
            print(f"Delivering '{message}' to {endpoint}")


if __name__ == "__main__":
    sns = SNSMock()
    print(sns.create_topic("alerts"))
    print(sns.subscribe("alerts", "email@example.com"))
    print(sns.subscribe("alerts", "sms:+15551234567"))
    sns.publish("alerts", "Server down!")

Output

stdout
arn:aws:sns:us-east-1:123456789012:alerts
subscription:alerts:email@example.com
subscription:alerts:sms:+15551234567
Delivering 'Server down!' to email@example.com
Delivering 'Server down!' to sms:+15551234567

How it works

This mock uses a dictionary keyed by topic name to store lists of subscriber endpoints, mirroring SNS topic fanout. create_topic adds an empty list if the topic is new and returns a realistic ARN string. subscribe appends the endpoint to the topic's list, and publish iterates over all subscribers, printing the delivery. This replaces SDK calls with clear in-process logic, making it easy to test or demonstrate concepts without network or AWS credentials.

Common mistakes

  • Forgetting to initialize the topic list before appending, causing KeyError.
  • Raising exceptions on publish to a non-existent topic instead of silently ignoring.
  • Printing delivery instead of returning results, which complicates unit testing.
  • Hard-coding ARNs without including a real account ID or region format.

Variations

  1. Use a `defaultdict(list)` to auto-create topic lists on first subscribe.
  2. Add a `delete_topic` method to remove topics and simulate cleanup.

Real-world use cases

  • Unit-testing application code that depends on SNS without incurring AWS costs.
  • Building a local development harness to simulate event-driven microservices.
  • Teaching pub/sub concepts in a classroom or training environment.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Cloud + Python

Related tutorials and quizzes for this topic.