Reference library

Streaming & messaging

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

4 matches

Batch Consume Process Commit Pattern in Python

A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.

streaming batch-processing queues

Python

import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    de…

15 0 Open

Streaming & messaging medium

How to Mock a Kafka Producer Batch Send in Python

Simulate a Kafka producer in Python that sends batched JSON events with mock partitions and latency for testing streaming pipelines without a real broker.

kafka mock streaming

Python

import json
import random
import time
from datetime import datetime


class MockKafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.sent_messages = []

    def send(self, value, key=None):
        message = {
            "topic": self.topic,
            "key": key,
            "value"…

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

Kafka Consumer Poll Loop Mock in Python

Simulate a Kafka consumer poll loop with a mock class, process messages in batches, and commit offsets to understand streaming consumption patterns.

kafka streaming mock

Python

import time

class MockKafkaConsumer:
    def __init__(self, topic, messages):
        self.topic = topic
        self.messages = list(messages)
        self.position = 0

    def poll(self, timeout_ms=100):
        if self.position >= len(self.messages):
            time.sleep(timeout_ms / 1000)
            return []…

13 0 Open

Browse by section

Each section groups closely related Python snippets.