Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
How to Build a Flow Control Credit Window in Python
A Python class that reserves, confirms, releases, and settles credit to limit message flow and prevent overload in streaming pipelines.
class CreditWindow:
def __init__(self, max_credit=1000):
self.max_credit = max_credit
self.used_credit = 0
self.pending_credit = 0
def try_reserve(self, amount):
available = self.max_credit - self.used_credit - self.pending_credit
if available >= amount:
…How to Mock a Kafka Producer Batch Send in Python
Simulate a Kafka producer in Python that sends batched JSON events with mock partitions and latency for testing streaming pipelines without a real broker.
import json
import random
import time
from datetime import datetime
class MockKafkaProducer:
def __init__(self, topic):
self.topic = topic
self.sent_messages = []
def send(self, value, key=None):
message = {
"topic": self.topic,
"key": key,
"value"…How to Stream Join Windowed Mock Topics in Python
Simulates two message topics and joins their events when timestamps fall within a sliding time window using Python generators and deques.
import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field
@dataclass
class Event:
key: str
value: int
timestamp: float = field(default_factory=time.time)
def generate_topic(prefix, keys, start_time):
while True:
yield Event(
…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…How to mock a CQRS projector read model update in Python
Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class OrderReadModel:
order_id: str
customer_name: str
total: float
status: str = "pending"
items: List[Dict] = field(default_factory=list)
def apply_event(self, event_type: str, payload: Dict) -> Non…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…In-Memory PubSub Topic Subscribe Mock in Python
Build a thread-safe in-memory publish/subscribe mock where handlers subscribe to named topics and receive every message published to them.
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)
def publish(self, topic, message):
for callback in self.topics.get(topic, []):
…Mock Watermark Late Event Side Output in Python
Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.
from datetime import datetime, timedelta
from typing import List, Tuple
def watermark_mock(
events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
"""Simulate watermarking: events arriving on time vs. late by ch…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.