Reference library

Streaming & messaging

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

5 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 medium

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.

redis streams consumer groups

Python

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…

12 0 Open

Streaming & messaging medium

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.

streaming join generator

Python

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

14 0 Open

Streaming & messaging medium

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.

session-window streaming timeout

Python

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…

13 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

Browse by section

Each section groups closely related Python snippets.