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
…Dead Letter Queue Failed Messages List Mock in Python
Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.
import json
from collections import deque
class Message:
def __init__(self, message_id, payload, attempts=0):
self.message_id = message_id
self.payload = payload
self.attempts = attempts
def __repr__(self):
return f"Message(id={self.message_id}, attempts={self.attempts})"
c…Dedupe processed message IDs in Python
Filters an inbox of messages by removing items whose IDs have already been processed, using a set for fast lookups.
from pathlib import Path
import json
def dedupe_processed_ids(inbox_file: Path, processed_file: Path) -> list:
processed = set(json.loads(processed_file.read_text()))
inbox = json.loads(inbox_file.read_text())
deduped = [item for item in inbox if item["id"] not in processed]
return deduped
if __nam…Event Envelope with Schema Version Field in Python
Build a typed event envelope dataclass with an explicit schema version field for mock streaming scenarios.
from dataclasses import dataclass, field
from datetime import datetime
import uuid
@dataclass
class Event:
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
event_type: str = "user.created"
version: str = "1.0.0"
created_at: str = field(default_factory=lambda: datetime.utcnow().isoform…Exactly Once Idempotent Consumer Store in Python
A mock key-value store that guarantees exactly-once processing by rejecting duplicate message keys in a message or event stream.
from collections import defaultdict
class ExactlyOnceStore:
def __init__(self):
self.processed = defaultdict(set)
self.data = {}
def consume(self, key, value):
if key in self.data:
return False
self.data[key] = value
return True
def get_processed_count…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 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 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 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 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…Implement a FIFO Message Queue in Python with deque
This code implements a FIFO (first-in-first-out) message queue class using Python's collections.deque, providing enqueue, dequeue, peek, and size operations.
from collections import deque
class MessageQueue:
def __init__(self):
self.queue = deque()
def enqueue(self, message):
self.queue.append(message)
print(f"Enqueued: {message}")
def dequeue(self):
if self.is_empty():
print("Queue is empty, cannot dequeue.")
…Mock NATS queue group load balancing in Python
Simulates a NATS queue group where each message is delivered to exactly one subscriber using random selection with a lightweight mock.
import random
import time
from collections import defaultdict
class MockQueueGroup:
"""Mock a NATS queue group: each message is delivered to exactly one subscriber."""
def __init__(self, subscribers):
self.subscribers = subscribers
def publish(self, message):
receiver = random.choice(se…Redis Pub/Sub Channel Subscribe Mock in Python
A lightweight in-memory mock of Redis pub/sub that lets you subscribe to channels, publish messages, and verify handler behavior in tests without a real Redis server.
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def subscribe(self, channel):
if channel not in self.channels:
self.channels[channel] = []
return self.channels[channel]
def publish(self, channel, message):
if channel in self.channels:
…Sliding Window Average with Deque in Python
Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.
from collections import deque
class SlidingAverage:
def __init__(self, window_size):
self.window_size = window_size
self.window = deque()
self.total = 0
def add(self, value):
self.window.append(value)
self.total += value
if len(self.window) > self.window_size:
…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.