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…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 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 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 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 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 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…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:
…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.