Streaming & messaging
Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.
How to deduplicate messages by ID in Python
Track seen message IDs in a set to skip duplicate messages and store unique content in a dict, with exact output showing which messages were added or skipped.
import time
class MessageStore:
def __init__(self):
self.seen_ids = set()
self.messages = {}
def add(self, message_id, content, timestamp=None):
timestamp = timestamp or time.time()
if message_id in self.seen_ids:
return False
self.seen_ids.add(message_…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.