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.
Python code
26 linesclass 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
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
- Use a `defaultdict(list)` to auto-create topic lists on first subscribe.
- 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
More from Cloud + Python
- Build a URL Shortener Client with Python medium
- Create a Cloud Storage Helper Class in Python easy
- Create a Data Helper Class for Beginners in Python easy
- Cross Account Role Chaining Mock Credentials in Python medium
- Exponential Backoff with Jitter for Cloud API Calls in Python medium
- Generate Mock CloudFormation Stack Events in Python easy
Keep learning
Related tutorials and quizzes for this topic.