Reference library

Streaming & messaging

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

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

How to mock a CQRS projector read model update in Python

Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.

cqrs projector read-model

Python

from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> Non…

11 0 Open

Browse by section

Each section groups closely related Python snippets.