Reference library

Streaming & messaging

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

4 matches

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

Streaming & messaging easy

How to Simulate a Micro-Batch Interval Trigger in Python

A dataclass-based mock that emits batch numbers at fixed intervals, mimicking a micro-batch streaming scheduler for testing and development.

streaming mock dataclass

Python

import time
from dataclasses import dataclass, field
from typing import List, Callable


@dataclass
class MicroBatchTriggerMock:
    batch_interval_seconds: float = 0.5
    max_batches: int = 5
    _batches_emitted: int = 0
    _next_emit_time: float = field(init=False, default=0)

    def start(self, on_batch: Callab…

13 0 Open

Streaming & messaging easy

How to Wrap Message Attributes in a CloudEvent with Python

Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.

cloudevents messaging dataclasses

Python

import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone


@dataclass
class CloudEvent:
    message_attributes: Dict[str, Any] = field(default_factory=dict)

    def wrap(self, event_id: str, source: str, event_type: str, data: Any):
        self…

13 0 Open

Streaming & messaging easy

Using the retained message flag in MQTT with Python

This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.

mqtt paho-mqtt iot

Python

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # Subscribe to a topic and check retained flag
    client.subscribe("test/retained")
    print("Subscribed to test/retained")

def on_message(client, userdata, msg):
    # msg.retain is the M…

14 0 Open

Browse by section

Each section groups closely related Python snippets.