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…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…How to Aggregate Periodic Snapshot Data in Python
Generates mock snapshot data and groups values into periods to compute average aggregates with Python's standard library.
import random
from collections import defaultdict
def snapshot_aggregate(n=10, period=3):
data = defaultdict(list)
for i in range(n):
key = f"item_{i % period}"
data[key].append(random.randint(1, 100))
return dict(data)
def aggregate_periodic(snapshots, period=3):
result = {}
for …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 Priority Queue for Messages in Python
Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.
import heapq
from dataclasses import dataclass, field
from typing import Any
@dataclass(order=True)
class Message:
priority: int
sequence: int = field(compare=False)
content: str = field(compare=False)
class PriorityQueue:
def __init__(self):
self._heap = []
def push(self, priority: int,…How to Read Redis Streams with XREADGROUP in Python
Read new messages from a Redis stream using a consumer group with XREADGROUP, handling JSON payloads and group creation.
import redis
import json
def read_group_messages(stream_key, group_name, consumer_name, count=10):
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
try:
r.xgroup_create(stream_key, group_name, id="0", mkstream=True)
except redis.exceptions.ResponseError:
pass
messag…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 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…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.