Reference library

Streaming & messaging

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

5 matches

How to Build a Materialized View Updater Consumer Mock in Python

A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.

import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


@dataclass
class MaterializedViewUpdater:
    """Mock updater that consumes change events and refreshes a view."""
    refresh: Optional[Callable[[str], None]] = None
    queue: Deque[tuple…

14 0 Open

Streaming & messaging medium

How to Mock NATS Subject Hierarchies with Wildcards in Python

Build a lightweight NATS-style pub/sub mock that matches subject hierarchies with '*' and '>' wildcards for tests or prototypes.

nats pubsub wildcards

Python

# Mock a simplified NATS subject hierarchy with wildcard matching
# Supports: exact match, '*' (single token), '>' (tail wildcard)

class NATSSubjectMock:
    def __init__(self):
        self.subscriptions = {}  # subject -> list of callbacks

    def subscribe(self, subject, callback):
        self.subscriptions.setd…

13 0 Open

Streaming & messaging medium

How to Mock Offset Commit Auto vs Manual in Python

Demonstrates a Kafka-style offset commit function with auto/manual modes and tests it using unittest.mock.patch.

unittest mocking kafka

Python

from unittest.mock import Mock, patch

def commit_offsets(topic_partition_offsets, auto_commit=False):
    """Manually commit offsets or simulate auto-commit."""
    if auto_commit:
        print(f"Auto-committing offsets: {topic_partition_offsets}")
        return {"status": "auto_committed"}
    
    print(f"Manuall…

15 0 Open

Streaming & messaging medium

How to Mock a Kafka Rebalance Listener in Python

Simulate Kafka consumer rebalance callbacks (on_partitions_revoked and on_partitions_assigned) with a mock consumer to test listener logic.

kafka rebalance mocking

Python

import time
from collections import defaultdict


class MockKafkaConsumer:
    def __init__(self):
        self.assignments = defaultdict(list)
        self.rebalances = 0

    def assign(self, partitions):
        self.rebalances += 1
        self.assignments.clear()
        for partition in partitions:
            s…

15 0 Open

Streaming & messaging medium

Mock Redis Streams XADD and XREAD in Python

A pure-Python mock of Redis streams that implements basic XADD, XREAD, and XLEN behavior for local testing without a real Redis server.

redis streams mocking

Python

import redis
import time
import threading


class MockRedisStreams:
    def __init__(self):
        self.streams = {}

    def xadd(self, stream_name, fields):
        if stream_name not in self.streams:
            self.streams[stream_name] = []
        entry_id = f"{time.time_ns()}-{len(self.streams[stream_name])}"
…

13 0 Open

Browse by section

Each section groups closely related Python snippets.