Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
Build a Streaming Messaging Helper in Python
Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.
from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time
@dataclass
class Message:
user: str
text: str
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().strftime("%H:%M:%S")
class…How to Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
from collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
…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 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 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 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 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 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…Implement a retry queue with visibility timeout in Python
This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.
import time
from collections import deque
class SimpleQueue:
def __init__(self, visibility_timeout=2):
self.queue = deque()
self.in_flight = {}
self.visibility_timeout = visibility_timeout
def send(self, message):
self.queue.append(message)
def receive(self):
if …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.