Reference library

Streaming & messaging

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

3 matches

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 Serialize and Deserialize JSON Event Payloads in Python

Define an EventPayload class with custom to_json and from_json methods to convert event objects to JSON strings and back, using datetime parsing.

json serialization datetime

Python

import json
from datetime import datetime


class EventPayload:
    def __init__(self, event_id, event_type, timestamp, data):
        self.event_id = event_id
        self.event_type = event_type
        self.timestamp = timestamp
        self.data = data

    def to_json(self):
        return json.dumps({
          …

12 0 Open

Streaming & messaging easy

Mock NATS queue group load balancing in Python

Simulates a NATS queue group where each message is delivered to exactly one subscriber using random selection with a lightweight mock.

nats queue-group messaging

Python

import random
import time
from collections import defaultdict


class MockQueueGroup:
    """Mock a NATS queue group: each message is delivered to exactly one subscriber."""

    def __init__(self, subscribers):
        self.subscribers = subscribers

    def publish(self, message):
        receiver = random.choice(se…

13 0 Open

Browse by section

Each section groups closely related Python snippets.