Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
At Most Once Fire-and-Forget Mock in Python
A Python mock that enforces send() is called at most once and records the arguments for verification.
class FireForgetMock:
def __init__(self):
self._calls = 0
self._last_args = None
self._last_kwargs = None
def send(self, *args, **kwargs):
if self._calls > 0:
raise RuntimeError("send() called more than once")
self._calls += 1
self._last_args = args
…How to Build a Materialized View Updater Consumer Mock in Python
A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional
@dataclass
class MaterializedViewUpdater:
"""Mock updater that consumes change events and refreshes a view."""
refresh: Optional[Callable[[str], None]] = None
queue: Deque[tuple…How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python
Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.
import time
class PubSub:
def __init__(self):
self.subscribers = {}
def subscribe(self, topic, callback):
if topic not in self.subscribers:
self.subscribers[topic] = []
self.subscribers[topic].append(callback)
def publish(self, topic, message):
if topic in sel…How to Implement a Tumbling Window Counter in Python
Count events that fall within a fixed-size sliding time window using a deque and pruning logic.
from collections import deque
import time
class TumblingWindowCounter:
def __init__(self, window_size_seconds):
self.window_size = window_size_seconds
self.window = deque()
def add_event(self, timestamp):
self.window.append(timestamp)
def count(self, current_time):
while…How to Implement an In-Memory Pub/Sub System in Python
This code implements a simple in-memory publish/subscribe system in Python, allowing topics, callbacks, and message broadcasting.
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)
return lambda: self.unsubscribe(topic, callback)
def unsubscribe(self, topic, callb…How to Mock a Kafka Rebalance Listener in Python
Simulate Kafka consumer rebalance callbacks (on_partitions_revoked and on_partitions_assigned) with a mock consumer to test listener logic.
import time
from collections import defaultdict
class MockKafkaConsumer:
def __init__(self):
self.assignments = defaultdict(list)
self.rebalances = 0
def assign(self, partitions):
self.rebalances += 1
self.assignments.clear()
for partition in partitions:
s…How to Stream Join Windowed Mock Topics in Python
Simulates two message topics and joins their events when timestamps fall within a sliding time window using Python generators and deques.
import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field
@dataclass
class Event:
key: str
value: int
timestamp: float = field(default_factory=time.time)
def generate_topic(prefix, keys, start_time):
while True:
yield Event(
…Implement a retry queue with visibility timeout in Python
This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.
import time
from collections import deque
class SimpleQueue:
def __init__(self, visibility_timeout=2):
self.queue = deque()
self.in_flight = {}
self.visibility_timeout = visibility_timeout
def send(self, message):
self.queue.append(message)
def receive(self):
if …Browse by section
Each section groups closely related Python snippets.
Streaming & messaging — Python code examples
What you will find here
This page collects streaming & messaging snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.