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 Implement Backpressure Pause Producer with a Bounded Queue in Python

Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.

queues backpressure threading

Python

import threading
import time
import queue
import random


class Producer:
    def __init__(self, q):
        self.q = q
        self.running = True

    def produce(self):
        while self.running:
            item = random.randint(1, 100)
            try:
                self.q.put(item, timeout=0.5)
              …

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

How to mock a CQRS projector read model update in Python

Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.

cqrs projector read-model

Python

from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> Non…

11 0 Open

Streaming & messaging medium

Mock Kafka Consumer Group Partition Assignment in Python

Simulates a Kafka consumer group's round-robin partition assignment with a Python class and prints assignments per consumer.

kafka consumer-group partition-assignment

Python

from collections import defaultdict


class ConsumerGroupAssignment:
    def __init__(self, group_name, topics_partitions):
        self.group_name = group_name
        self.consumers = {}
        self.assignments = defaultdict(set)
        topics_partitions = sorted(
            [(topic, partition) for topic, partiti…

14 0 Open

Browse by section

Each section groups closely related Python snippets.