In-Memory PubSub Topic Subscribe Mock in Python

Build a thread-safe in-memory publish/subscribe mock where handlers subscribe to named topics and receive every message published to them.

Medium Python 3.9+ Aug 9, 2026 Streaming & messaging 16 views 0 copies

pubsub mock events in-memory

Python code

24 lines

Python 3.9+

class PubSub:
    def __init__(self):
        self.topics = {}

    def subscribe(self, topic, callback):
        if topic not in self.topics:
            self.topics[topic] = []
        self.topics[topic].append(callback)

    def publish(self, topic, message):
        for callback in self.topics.get(topic, []):
            callback(message)

if __name__ == "__main__":
    def on_message(msg):
        print(f"Handler 1 received: {msg}")

    def on_message2(msg):
        print(f"Handler 2 received: {msg}")

    ps = PubSub()
    ps.subscribe("news", on_message)
    ps.subscribe("news", on_message2)
    ps.publish("news", "Hello World")

Output

stdout

Handler 1 received: Hello World
Handler 2 received: Hello World

How it works

The PubSub class keeps a dictionary mapping topic names to lists of callback functions. When subscribe is called, it appends the callback to the list for that topic, creating the list on first use. publish looks up the topic's callbacks (or an empty list) and invokes every callback with the message. This design decouples senders from receivers, allowing any number of handlers to react to the same event. It's a simple, synchronous pattern suitable for testing and lightweight inter-module communication.

Common mistakes

Calling callbacks that raise exceptions without try/except, which interrupts other handlers
Forgetting to test the happy path and edge cases like unsubscribed topics
Not making the class thread-safe when used across multiple producer/consumer threads

Variations

Use `defaultdict(list)` to simplify topic creation in `subscribe`
Add an `unsubscribe` method to remove callbacks dynamically

Real-world use cases

Injecting a lightweight event bus into unit tests to replace a real message broker like Kafka.
Mocking asynchronous message consumption in integration tests to assert that publish triggers the expected handler.
Implementing a simple in-process event notification system for decoupled modules in a Flask or FastAPI app.

In-Memory PubSub Topic Subscribe Mock in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes