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
…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…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 an Outbox Table Poll Publisher in Python
This code simulates an outbox pattern with a class that polls for pending records and publishes them as JSON messages, removing only those that are due.
import time
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
@dataclass
class OutboxRecord:
id: int
topic: str
payload: dict
created_at: datetime
class OutboxPollPublisher:
def __init__(self, poll_interval_seconds=1):
self.poll_interval = poll…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 Track Session Windows with Gap Timeout in Python
A Python class that groups events into sessions, closing a session when the gap between events exceeds a timeout threshold.
import time
class SessionWindow:
"""Track sessions with a gap timeout (mock)."""
def __init__(self, timeout_seconds=5):
self.timeout = timeout_seconds
self.session_start = None
self.last_event_time = None
self.event_count = 0
self.events = []
def add_event…Simulate RabbitMQ QoS Prefetch Count in Python
Mocks RabbitMQ QoS prefetch semantics using threading and a queue to cap concurrent unacked message processing per worker.
import threading
import time
import queue
class RabbitMQMock:
def __init__(self, prefetch_count=1):
self.prefetch_count = prefetch_count
self.channel_queue = queue.Queue()
self.currently_processing = 0
self.lock = threading.Lock()
def start_consuming(self, messages, worker_co…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.