Reference library

Streaming & messaging

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

6 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…

14 0 Open

Streaming & messaging easy

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.

dataclasses deque mocking

Python

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 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 easy

How to Mock RabbitMQ Ack Nack Requeue in Python

A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.

rabbitmq testing mock

Python

import json
from collections import deque


class MockChannel:
    def __init__(self):
        self.acked = []
        self.nacked = []
        self.requeued = []

    def basic_ack(self, delivery_tag):
        self.acked.append(delivery_tag)

    def basic_nack(self, delivery_tag, requeue=False):
        self.nacked.…

15 0 Open

Streaming & messaging medium

How to Simulate RabbitMQ Exchange Routing in Python

Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.

rabbitmq routing messaging

Python

from collections import defaultdict

def route_message(exchanges, exchange_name, routing_key):
    """
    Simulate RabbitMQ exchange routing using a nested dict structure.
    Returns list of queue names that match the routing key.
    """
    queues = exchanges.get(exchange_name, {})
    matched = []
    
    for pa…

14 0 Open

Streaming & messaging easy

How to mock RabbitMQ queue binding with routing keys in Python

A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.

rabbitmq messaging pika

Python

import pika
import sys


def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
    """
    Mock RabbitMQ queue binding with routing keys.
    Prints the binding configuration instead of connecting to a real broker.
    """
    for routing_key in routing_keys:
        binding = {
            "q…

14 0 Open

Browse by section

Each section groups closely related Python snippets.