Reference library

Streaming & messaging

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

5 matches

At Most Once Fire-and-Forget Mock in Python

A Python mock that enforces send() is called at most once and records the arguments for verification.

class FireForgetMock:
    def __init__(self):
        self._calls = 0
        self._last_args = None
        self._last_kwargs = None

    def send(self, *args, **kwargs):
        if self._calls > 0:
            raise RuntimeError("send() called more than once")
        self._calls += 1
        self._last_args = args
…

17 0 Open

Streaming & messaging easy

Dedupe processed message IDs in Python

Filters an inbox of messages by removing items whose IDs have already been processed, using a set for fast lookups.

deduplication streaming json

Python

from pathlib import Path
import json


def dedupe_processed_ids(inbox_file: Path, processed_file: Path) -> list:
    processed = set(json.loads(processed_file.read_text()))
    inbox = json.loads(inbox_file.read_text())
    deduped = [item for item in inbox if item["id"] not in processed]
    return deduped


if __nam…

13 0 Open

Streaming & messaging easy

Event Envelope with Schema Version Field in Python

Build a typed event envelope dataclass with an explicit schema version field for mock streaming scenarios.

event dataclass messaging

Python

from dataclasses import dataclass, field
from datetime import datetime
import uuid


@dataclass
class Event:
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    event_type: str = "user.created"
    version: str = "1.0.0"
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoform…

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

How to Mock MQTT Topic Subscriptions with QoS in Python

Build a lightweight MQTT client mock that tracks topic subscriptions with QoS levels and simulates wildcard message delivery.

mqtt mock qos

Python

import time
from collections import defaultdict

class MockMQTTClient:
    def __init__(self):
        self.subscriptions = defaultdict(list)
        self.messages = []
    
    def subscribe(self, topic, qos=0):
        self.subscriptions[topic].append(qos)
        print(f"Subscribed to '{topic}' with QoS {qos}")
   …

14 0 Open

Browse by section

Each section groups closely related Python snippets.