Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
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 Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
from collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
…How to Build a Mock Change Data Capture Event Stream in Python
Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.
from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta
seed(42) # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)
def mock_cdc_event(stream_index: int) -> dict:
…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 Priority Queue for Messages in Python
Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.
import heapq
from dataclasses import dataclass, field
from typing import Any
@dataclass(order=True)
class Message:
priority: int
sequence: int = field(compare=False)
content: str = field(compare=False)
class PriorityQueue:
def __init__(self):
self._heap = []
def push(self, priority: int,…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 Kafka Topic Partitions with a Python dict of lists
Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.
from collections import defaultdict
class KafkaTopicPartitionMock:
"""A simple mock for Kafka topic-partition assignment using dict of lists."""
def __init__(self, topic):
self.topic = topic
self.partitions = defaultdict(list) # partition_id -> list of messages
def produce(self, message…How to Mock MQTT Topic Subscriptions with QoS in Python
Build a lightweight MQTT client mock that tracks topic subscriptions with QoS levels and simulates wildcard message delivery.
import time
from collections import defaultdict
class MockMQTTClient:
def __init__(self):
self.subscriptions = defaultdict(list)
self.messages = []
def subscribe(self, topic, qos=0):
self.subscriptions[topic].append(qos)
print(f"Subscribed to '{topic}' with QoS {qos}")
…How to Mock RabbitMQ Ack Nack Requeue in Python
A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.
import json
from collections import deque
class MockChannel:
def __init__(self):
self.acked = []
self.nacked = []
self.requeued = []
def basic_ack(self, delivery_tag):
self.acked.append(delivery_tag)
def basic_nack(self, delivery_tag, requeue=False):
self.nacked.…How to Partition and Order Kafka-Style Messages by Key in Python
Group messages with the same key into ordered buckets using hashing and a defaultdict, mimicking Kafka partition ordering.
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class Message:
key: str
content: str
def partition_and_order(messages, num_partitions=3):
partitions = defaultdict(list)
for msg in messages:
partition_id = hash(msg.key) % num_partitions
partitions[parti…How to Serialize and Deserialize JSON Event Payloads in Python
Define an EventPayload class with custom to_json and from_json methods to convert event objects to JSON strings and back, using datetime parsing.
import json
from datetime import datetime
class EventPayload:
def __init__(self, event_id, event_type, timestamp, data):
self.event_id = event_id
self.event_type = event_type
self.timestamp = timestamp
self.data = data
def to_json(self):
return json.dumps({
…How to Simulate a Micro-Batch Interval Trigger in Python
A dataclass-based mock that emits batch numbers at fixed intervals, mimicking a micro-batch streaming scheduler for testing and development.
import time
from dataclasses import dataclass, field
from typing import List, Callable
@dataclass
class MicroBatchTriggerMock:
batch_interval_seconds: float = 0.5
max_batches: int = 5
_batches_emitted: int = 0
_next_emit_time: float = field(init=False, default=0)
def start(self, on_batch: Callab…How to Wrap Message Attributes in a CloudEvent with Python
Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.
import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone
@dataclass
class CloudEvent:
message_attributes: Dict[str, Any] = field(default_factory=dict)
def wrap(self, event_id: str, source: str, event_type: str, data: Any):
self…How to deduplicate messages by ID in Python
Track seen message IDs in a set to skip duplicate messages and store unique content in a dict, with exact output showing which messages were added or skipped.
import time
class MessageStore:
def __init__(self):
self.seen_ids = set()
self.messages = {}
def add(self, message_id, content, timestamp=None):
timestamp = timestamp or time.time()
if message_id in self.seen_ids:
return False
self.seen_ids.add(message_…How to mock RabbitMQ queue binding with routing keys in Python
A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.
import pika
import sys
def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
"""
Mock RabbitMQ queue binding with routing keys.
Prints the binding configuration instead of connecting to a real broker.
"""
for routing_key in routing_keys:
binding = {
"q…Using the retained message flag in MQTT with Python
This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.
import paho.mqtt.client as mqtt
def on_connect(client, userdata, flags, rc):
print(f"Connected with result code {rc}")
# Subscribe to a topic and check retained flag
client.subscribe("test/retained")
print("Subscribed to test/retained")
def on_message(client, userdata, msg):
# msg.retain is the M…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.