Reference library

Streaming & messaging

Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.

4 matches

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.

aggregation snapshots streaming

Python

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 …

14 0 Open

Streaming & messaging easy

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.

deduplication set messaging

Python

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_…

12 0 Open

Streaming & messaging medium

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.

watermark streaming side output

Python

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…

11 0 Open

Streaming & messaging easy

Sliding Window Average with Deque in Python

Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.

sliding-window deque streaming

Python

from collections import deque

class SlidingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque()
        self.total = 0

    def add(self, value):
        self.window.append(value)
        self.total += value
        if len(self.window) > self.window_size:
…

13 0 Open

Browse by section

Each section groups closely related Python snippets.