Reference library

Streaming & messaging

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

5 matches

Dead Letter Queue Failed Messages List Mock in Python

Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.

dead-letter-queue messaging retry

Python

import json
from collections import deque


class Message:
    def __init__(self, message_id, payload, attempts=0):
        self.message_id = message_id
        self.payload = payload
        self.attempts = attempts

    def __repr__(self):
        return f"Message(id={self.message_id}, attempts={self.attempts})"


c…

15 0 Open

Streaming & messaging medium

How to Build a Flow Control Credit Window in Python

A Python class that reserves, confirms, releases, and settles credit to limit message flow and prevent overload in streaming pipelines.

flow-control credit-window streaming

Python

class CreditWindow:
    def __init__(self, max_credit=1000):
        self.max_credit = max_credit
        self.used_credit = 0
        self.pending_credit = 0
    
    def try_reserve(self, amount):
        available = self.max_credit - self.used_credit - self.pending_credit
        if available >= amount:
           …

14 0 Open

Streaming & messaging easy

How to Build a Mock Change Data Capture Event Stream in Python

Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.

cdc mock event-stream

Python

from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta

seed(42)  # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)

def mock_cdc_event(stream_index: int) -> dict:
…

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

Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

watermark streaming side output

Python

from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by ch…

11 0 Open

Browse by section

Each section groups closely related Python snippets.