Reference library

Streaming & messaging

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

3 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 Mock Kafka Topic Partitions with a Python dict of lists

Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.

kafka mock partitions

Python

from collections import defaultdict

class KafkaTopicPartitionMock:
    """A simple mock for Kafka topic-partition assignment using dict of lists."""

    def __init__(self, topic):
        self.topic = topic
        self.partitions = defaultdict(list)  # partition_id -> list of messages

    def produce(self, message…

15 0 Open

Streaming & messaging easy

How to Partition and Order Kafka-Style Messages by Key in Python

Group messages with the same key into ordered buckets using hashing and a defaultdict, mimicking Kafka partition ordering.

streaming partitioning kafka-pattern

Python

from dataclasses import dataclass
from collections import defaultdict

@dataclass
class Message:
    key: str
    content: str

def partition_and_order(messages, num_partitions=3):
    partitions = defaultdict(list)
    for msg in messages:
        partition_id = hash(msg.key) % num_partitions
        partitions[parti…

14 0 Open

Browse by section

Each section groups closely related Python snippets.