Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
Batch Consume Process Commit Pattern in Python
A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.
import random
import threading
import time
from collections import deque
class MockBatchProcessor:
def __init__(self, process_func, commit_func, batch_size=5):
self.queue = deque()
self.batch_size = batch_size
self.process_func = process_func
self.commit_func = commit_func
de…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 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 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 Simulate RabbitMQ Exchange Routing in Python
Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.
from collections import defaultdict
def route_message(exchanges, exchange_name, routing_key):
"""
Simulate RabbitMQ exchange routing using a nested dict structure.
Returns list of queue names that match the routing key.
"""
queues = exchanges.get(exchange_name, {})
matched = []
for pa…Implement the Transactional Outbox Pattern with SQLite in Python
A Python implementation of the transactional outbox pattern using SQLite, ensuring atomic writes of order data and outbox events in a single transaction while supporting reliable message publishing and consumption.
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
import json
@dataclass
class Order:
order_id: str
amount: float
status: str
class TransactionalOutbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self._create_tab…Kafka Consumer Poll Loop Mock in Python
Simulate a Kafka consumer poll loop with a mock class, process messages in batches, and commit offsets to understand streaming consumption patterns.
import time
class MockKafkaConsumer:
def __init__(self, topic, messages):
self.topic = topic
self.messages = list(messages)
self.position = 0
def poll(self, timeout_ms=100):
if self.position >= len(self.messages):
time.sleep(timeout_ms / 1000)
return []…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.